package com.shujia.bigdata.word;

import java.io.*;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Reduce {
    public static void main(String[] args) {

        //创建线程池
        ExecutorService threadPool = Executors.newFixedThreadPool(4);

        File pfile = new File("java/data/word_tmp2");

        File[] files = pfile.listFiles();

        if (files == null) {
            return;
        }

        //将相同后缀结尾的文件放到同一个集合中
        HashMap<String, ArrayList<File>> listHashMap = new HashMap<>();

        for (File file : files) {
            String partition = file.getName().split("-")[1];
            ArrayList<File> files1 = listHashMap.get(partition);

            if (files1 == null) {
                files1 = new ArrayList<>();
                files1.add(file);
                listHashMap.put(partition, files1);
            } else {
                files1.add(file);
                //覆盖之前的集合
                listHashMap.put(partition, files1);
            }
        }

        for (String partition : listHashMap.keySet()) {

            ArrayList<File> files1 = listHashMap.get(partition);

            ReduceTask reduceTask = new ReduceTask(files1, partition);

            //启动线程
            threadPool.submit(reduceTask);
        }


        //关闭线程池
        threadPool.shutdown();


    }
}

/**
 * 每一个线程处理同一个后缀结尾的文件
 */

class ReduceTask extends Thread {

    private ArrayList<File> files;
    private String partition;

    public ReduceTask(ArrayList<File> files, String partition) {
        this.files = files;
        this.partition = partition;
    }

    @Override
    public void run() {

        HashMap<String, Long> map = new HashMap<>();

        //处理所有文件
        for (File file : files) {

            try {
                FileReader fileReader = new FileReader(file);

                BufferedReader bufferedReader = new BufferedReader(fileReader);

                String line;
                while ((line = bufferedReader.readLine()) != null) {
                    String[] split = line.split(",");

                    String word = split[0];
                    Long count = Long.parseLong(split[1]);

                    Long aLong = map.get(word);//如果不存在返回null

                    //汇总所有文件单词的数量
                    if (aLong == null) {
                        map.put(word, count);
                    } else {
                        map.put(word, count + aLong);
                    }
                }

                bufferedReader.close();
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        /**
         * 将汇总结果写入文件
         *
         * 每个reduce都会产生一个文件
         *
         */

        try {
            FileWriter fileWriter = new FileWriter("java/data/word_result/" + partition);
            BufferedWriter bufferedWriter = new BufferedWriter(fileWriter);
            for (String word : map.keySet()) {

                Long count = map.get(word);

                String line = word + "," + count;

                bufferedWriter.write(line);
                bufferedWriter.newLine();
            }

            bufferedWriter.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
